package plagiarismChecker;

import java.io.BufferedReader;
import java.io.DataInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.math.BigDecimal;
import java.math.MathContext;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

import map.MapJob1;
import map.MapJob2;
import map.MapJob3;
import map.MapJob4;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.log4j.Level;
import org.apache.log4j.LogManager;

import reduce.ReduceJob1;
import reduce.ReduceJob2;
import reduce.ReduceJob3;
import reduce.ReduceJob4;

public class MapReduceSkeleton {

	public static String separator = "#########";

	static Configuration conf;
	
	public static String comparableFile;

    public static void main(String[] args) throws Exception {
    	LogManager.getRootLogger().setLevel((Level)Level.ERROR);
    	conf = new Configuration();

    	if(args.length < 2) {
    		System.out.println("Too few arguments! Needs comparable_file comparable_folder");
    		System.out.println("Excample: comparable_file.txt comparable_folder");
    		System.exit(0);
    	}else if(args.length == 3){
    		separator = args[2];
    	}

    	String pathin = "tmp_data/input.txt";
    	String pathout = "tmp_data/output";
    	deleteDir(new File(pathout));
    	
    	comparableFile = args[0];
    	String comparableFolder = args[1];

    	conf.setStrings("comparable_file", comparableFile);
    	conf.setStrings("comparable_folder", comparableFolder);
    	
    	conf.setInt("nr_of_files_in_folder", getFilesInFolder(comparableFolder).size());

    	List<File> files= getFilesInFolder(comparableFolder);
    	for (File fileName : files) {
    		DataReader.writeData(fileName.toString(), "1", files.size(), pathin, conf);
    	}
    	
    	Job job1 = new Job(conf);
		job1.setJarByClass(MapReduceSkeleton.class);

		job1.setMapperClass(MapJob1.class);
		job1.setReducerClass(ReduceJob1.class);

		job1.setInputFormatClass(SequenceFileInputFormat.class);
		job1.setOutputFormatClass(SequenceFileOutputFormat.class);
		
		job1.setMapOutputKeyClass(Text.class);
		job1.setMapOutputValueClass(Text.class);

		job1.setOutputKeyClass(Text.class);
		job1.setOutputValueClass(Text.class);

		FileInputFormat.addInputPath(job1, new Path(pathin));
		FileOutputFormat.setOutputPath(job1, new Path(pathout + "/job1"));
		job1.waitForCompletion(true);
		
		
		if (job1.isSuccessful()) {
			Job job2 = new Job(conf);
			job2.setJarByClass(MapReduceSkeleton.class);

			job2.setMapperClass(MapJob2.class);
			job2.setReducerClass(ReduceJob2.class);
			
			job2.setInputFormatClass(SequenceFileInputFormat.class);
			job2.setOutputFormatClass(SequenceFileOutputFormat.class);
			
			job2.setMapOutputKeyClass(Text.class);
			job2.setMapOutputValueClass(Text.class);

			job2.setOutputKeyClass(Text.class);
			job2.setOutputValueClass(Text.class);

			FileInputFormat.addInputPath(job2, new Path(pathout + "/job1"));
			FileOutputFormat.setOutputPath(job2, new Path(pathout + "/job2"));
			job2.waitForCompletion(true);
			
			if (job2.isSuccessful()) {
				Job job3 = new Job(conf);
				job3.setJarByClass(MapReduceSkeleton.class);

				job3.setMapperClass(MapJob3.class);
				job3.setReducerClass(ReduceJob3.class);
				
				job3.setInputFormatClass(SequenceFileInputFormat.class);
				job3.setOutputFormatClass(SequenceFileOutputFormat.class);
				
				job3.setMapOutputKeyClass(Text.class);
				job3.setMapOutputValueClass(Text.class);

				job3.setOutputKeyClass(Text.class);
				job3.setOutputValueClass(Text.class);

				FileInputFormat.addInputPath(job3, new Path(pathout + "/job2"));
				FileOutputFormat.setOutputPath(job3, new Path(pathout + "/job3"));
				
				job3.waitForCompletion(true);
				
				
				if (job3.isSuccessful()) {
					Job job4 = new Job(conf);
					job4.setJarByClass(MapReduceSkeleton.class);

					job4.setMapperClass(MapJob4.class);
					job4.setReducerClass(ReduceJob4.class);
					
					job4.setInputFormatClass(SequenceFileInputFormat.class);
					job4.setOutputFormatClass(SequenceFileOutputFormat.class);
					
					job4.setMapOutputKeyClass(Text.class);
					job4.setMapOutputValueClass(Text.class);

					job4.setOutputKeyClass(Text.class);
					job4.setOutputValueClass(Text.class);

					FileInputFormat.addInputPath(job4, new Path(pathout + "/job3"));
					FileOutputFormat.setOutputPath(job4, new Path(pathout + "/job4"));
					
					job4.waitForCompletion(true);
				}
				
			}
			
		}
		job1.waitForCompletion(true);
		
		/**
		 *  reads last output 
		 *  identical document are 0 and different 1,57 = PI/2
		 */
		ArrayList<HashMap<String, String>> lastResult = DataReader.readData(pathout + "/job4", conf);
		for (HashMap<String, String> resultLine : lastResult) {
			double vectorDistance = new BigDecimal(resultLine.get("value")).round(new MathContext(10)).doubleValue();
			double inversedPercentage = 100 - Math.round(vectorDistance / (Math.PI / 2) * 100);
			System.out.print(
				resultLine.get("key").replace(MapReduceSkeleton.separator, " -> ") + " "
			);
			if (vectorDistance <= 0.7227342478) {
				System.out.println("More or equal than 75% similarity. Vector distance is: " + vectorDistance);
			}
			else if (vectorDistance <= 1.047197551) {
				System.out.println("More or equal than 50% similarity. Vector distance is: " + vectorDistance);
			}
			else {
				System.out.println("Less than 50% similarity. Vector distance is: " + vectorDistance);
			}
		}
    }

    /**
     * Reads given folder and gets all files as objects
     * 
     * @param folderPath
     * @return files
     */
	public static List<File> getFilesInFolder(String folderPath) {
		List<File> files = null;
		try {
			File directory = new File(folderPath);
			files = new ArrayList<File>();
			for (File file : directory.listFiles()) {
				files.add(file);
			}
			files.add(new File(comparableFile));
		} catch (Exception e) {
			System.out.println("Invalid path for comparing file or folder");
			System.exit(0);
		}
		return files;
    }
    
    /**
     * Reads given file content
     * 
     * @param filePath
     * @return file content
     */
	public static String readFileContent(String filePath) {
    	File f = new File(filePath);
    	String fileContent = "";
    	try {
			FileInputStream fstream = new FileInputStream(f);
			DataInputStream in = new DataInputStream(fstream);
			BufferedReader br = new BufferedReader(new InputStreamReader(in));
			String strLine;
			while ((strLine = br.readLine()) != null) {
				fileContent += strLine;
			}
			in.close();
		}
		catch (Exception e) {
			System.err.println("Error: " + e.getMessage());
			System.exit(0);
		}
		return fileContent;
	}

	/**
	 * Reads folder files prints them with their content
	 * 
	 * @param folderPath
	 */
	public static void readFolderContent(String folderPath) {
		File directory = new File(folderPath);
		
		File files[] = directory.listFiles();

		for (File f : files) {
			System.out.println(f);
			try {
				// Open the file that is the first
				// command line parameter
				FileInputStream fstream = new FileInputStream(f);
				// Get the object of DataInputStream
				DataInputStream in = new DataInputStream(fstream);
				BufferedReader br = new BufferedReader(new InputStreamReader(in));
				String strLine;
				// Read File Line By Line
				while ((strLine = br.readLine()) != null) {
					// Print the content on the console
					System.out.println(strLine);
				}
				// Close the input stream
				in.close();
			} catch (Exception e) {// Catch exception if any
				System.err.println("Error: " + e.getMessage());
			}
		}
	}
	
	public static boolean deleteDir(File dir) {
	    if (dir.isDirectory()) {
	        String[] children = dir.list();
	        for (int i=0; i<children.length; i++) {
	            boolean success = deleteDir(new File(dir, children[i]));
	            if (!success) {
	                return false;
	            }
	        }
	    }

	    // The directory is now empty so delete it
	    return dir.delete();
	}
}